package com.seal.commons.infrastructure.rocketmq;

import com.seal.commons.infrastructure.facade.jms.MessageListener;
import com.seal.commons.infrastructure.facade.jms.annotation.JmsListener;
import com.seal.commons.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.annotation.AnnotationUtils;

import java.util.Iterator;
import java.util.Map;
import java.util.Set;

/**
 * ClassName:RocketmqListenerAutoExporter <br/>
 * Func:  rocketMQ自动订阅 <br/>
 * Date:     2019-06-06 10:48 <br/>
 *
 * @author yong.huang
 * @see
 * @since JDK 1.8
 */

@Slf4j
public class RocketMQListenerAutoExporter implements InitializingBean, ApplicationContextAware {

    private Map<String, MessageListener> listenerMap;

    private DefaultMQPushConsumer broadcastingConsumer;

    private DefaultMQPushConsumer clusteringConsumer;

    private ApplicationContext applicationContext;

    public RocketMQListenerAutoExporter(Map<String,MessageListener> listenerMap,
                                        DefaultMQPushConsumer broadcastingConsumer,
                                        DefaultMQPushConsumer clusteringConsumer){
        this.listenerMap = listenerMap;
        this.broadcastingConsumer = broadcastingConsumer;
        this.clusteringConsumer = clusteringConsumer;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        initListenerMap();
        subscribeBroadcasting();
        subscribeClustering();
    }

    public void subscribeBroadcasting() throws Exception {
        log.info("JmsListener subscribeTopic start ... ");
        Set<String> topics = listenerMap.keySet();
        for(String topic : topics) {
            MessageListener listener = listenerMap.get(topic);
            JmsListener jmsListener = AnnotationUtils.findAnnotation(listener.getClass(), JmsListener.class);
            if(MessageModel.BROADCASTING.name().equals(jmsListener.messageModel())){
                broadcastingConsumer.subscribe(jmsListener.topic(), jmsListener.tags());
                log.info("广播 subscribe broadcastingConsumer topic:[{}] tags:[{}]", topic, jmsListener.tags());
            }
        }
        log.info("广播  subscribeTopic end ... ");
    }

    public void subscribeClustering() throws Exception {
        log.info("JmsListener subscribeTopic start ... ");
        Set<String> topics = listenerMap.keySet();
        for(String topic : topics) {
            MessageListener listener = listenerMap.get(topic);
            JmsListener jmsListener = AnnotationUtils.findAnnotation(listener.getClass(), JmsListener.class);
            if(MessageModel.CLUSTERING.name().equals(jmsListener.messageModel())){
                clusteringConsumer.subscribe(jmsListener.topic(), jmsListener.tags());
                log.info("集群 subscribe broadcastingConsumer topic:[{}] tags:[{}]", topic, jmsListener.tags());
            }
        }
        log.info("集群 JmsListener subscribeTopic end ... ");
    }


    private void initListenerMap() throws Exception {
        DefaultListableBeanFactory beanFactory = (DefaultListableBeanFactory) applicationContext.getAutowireCapableBeanFactory();
        Map<String, Object> beans = applicationContext.getBeansWithAnnotation(JmsListener.class);
        Iterator<Map.Entry<String, Object>> it = beans.entrySet().iterator();
        while (it.hasNext()) {
            Map.Entry<String, Object> entry = it.next();
            if (!(entry.getValue() instanceof MessageListener)) {
                throw new Exception();
            }
            MessageListener listener = ((MessageListener) entry.getValue());
            JmsListener jmsListener = AnnotationUtils.findAnnotation(listener.getClass(), JmsListener.class);
            String topic =  beanFactory.resolveEmbeddedValue(jmsListener.topic());
            String tags = StringUtils.isEmpty(jmsListener.tags()) ? "default" : beanFactory.resolveEmbeddedValue(jmsListener.tags());
            listenerMap.put(topic +"_"+ tags, listener);
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
